package top.jolyoulu.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.io.{BufferedReader, InputStream, InputStreamReader}
import java.net.{ServerSocket, Socket}
import scala.Byte
import scala.collection.mutable
import scala.util.Random

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_DIY {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    //添加自定义采集器
    val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new ScoketReceiver())
    messageDS.print()

    ssc.start()
    ssc.awaitTermination()
  }

  //自定义数据采集器
  //继承Receiver[泛型](数据存储位置)
  class ScoketReceiver() extends Receiver[String](StorageLevel.MEMORY_ONLY) {

    private var flag = true;

    //启动时逻辑
    override def onStart(): Unit = {
      new Thread(new Runnable {
        override def run(): Unit = {
          while (flag){
            val message = "采集的数据为：" + new Random().nextInt(10).toString
            store(message)
            Thread.sleep(500)
          }
        }
      }).start()
    }

    override def onStop(): Unit = {
      flag = false
    }
  }
}
